package com.learn.datasource.custom;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;
import java.util.UUID;

/**
 * @create: 2023-04-18 09:49
 * @author: Mr.Du
 * --------------
 * @notes: 非并行自定义数据源
 **/
public class CustomSourceWithoutParallelDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<Order> mySourceFunction = env.addSource(new MySourceFunction());

        mySourceFunction.print();

        env.execute();
    }
}

class MySourceFunction implements SourceFunction<Order> {
    //固定写法，用于标识是否产生数据（关闭循环的标记）
    private boolean isRun = true;

    @Override
    public void run(SourceContext<Order> sctx) throws Exception {
        Random random = new Random();
        while(isRun) {
            String id = UUID.randomUUID().toString();
            String userId = random.nextInt(99) + "";
            int money = random.nextInt(999);
            long time = System.currentTimeMillis();

            sctx.collect(new Order(id,userId,money,time));
            Thread.sleep(1000L);

        }
    }

    @Override
    public void cancel() {
        this.isRun = false;
    }
}

@Data
@AllArgsConstructor
@NoArgsConstructor
class Order {
    //订单id
    private String id;
    //用户id
    private String userId;
    //订单金额
    private int money;
    //时间戳
    private Long timestamp;
}
